package com.nuonuo.queue;

import com.nuonuo.config.RedissonUtils;
import com.nuonuo.listener.RedisDelayedQueueListener;
import com.nuonuo.thread.QueueListenerThread;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

/**
 * @author duan
 * @version 1.0
 * @date 2024/1/25 16:39
 */
@Component
public class RedisDelayedQueueInit implements ApplicationContextAware {

    @Resource
    RedissonUtils redissonUtils;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String, RedisDelayedQueueListener> map = applicationContext.getBeansOfType(
                RedisDelayedQueueListener.class);
        for (Map.Entry<String, RedisDelayedQueueListener> taskEventListenerEntry : map.entrySet()) {
            String listenerName = taskEventListenerEntry.getValue().getClass().getName();
            startThread(listenerName, taskEventListenerEntry.getValue());
        }
    }


    /**
     * 描述: 启动线程获取队列
     *
     * @param queueName
     * @param redisDelayedQueueListener
     * @return void
     * @author liufan
     * @date 2023-03-30
     */
    private <T> void startThread(String queueName,
                                 RedisDelayedQueueListener<T> redisDelayedQueueListener) {
        RedissonClient redissonClient = redissonUtils.getRedissonClient();
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
        //服务重启后，无offer，take不到信息。
        redissonClient.getDelayedQueue(blockingFairQueue);
        QueueListenerThread queueListenerThread = new QueueListenerThread(blockingFairQueue, queueName, redisDelayedQueueListener);
        queueListenerThread.start();
    }
}
